/**
 * 
 */
package stc.skymobi.transport;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author hp
 *
 */
public class DefaultEndpoint extends AbstractEndpoint implements MutableEndpoint {
    private static final Logger logger = 
    	LoggerFactory.getLogger(DefaultEndpoint.class);

    private ExecutorService exec = 
        Executors.newSingleThreadExecutor();
    
    private	long			waitTimeout = 1000;
    
    private void doSendPending() {
		exec.submit(new Runnable() {

			public void run() {
				sendPending();
			}} );
    }
    
    private	void sendPending() {
    	try {
	    	if ( null == session ) {
	    		Thread.sleep(waitTimeout);	// sleep 1s 
	    	}
	    	else {
				Object bean = null, sharedBean = null;
				
				//	first check local pendings queue
				if ( null != pendings ) {
					//bean = pendings.poll(waitTimeout, TimeUnit.MILLISECONDS);
					bean = pendings.poll();
				}
	    		if ( null != sharedPendings) {
	    			//	check sharedPendings
	    			//sharedBean = sharedPendings.poll(waitTimeout, TimeUnit.MILLISECONDS);
	    			sharedBean = sharedPendings.poll();
	    		}
	    		if ( null != bean ) {
	    			session.write(bean);
	    		}
	    		if ( null != sharedBean ) {
	    			session.write(sharedBean);
	    		}
	    		if ( null == bean && null == sharedBean ) {
		    		Thread.sleep(waitTimeout);	// sleep 1s 
	    		}
	    	}
    	} catch (InterruptedException e) {
    		logger.error("sendPending", e);
		}
    	finally {
    		doSendPending();
    	}
    }
    
	public long getWaitTimeout() {
		return waitTimeout;
	}

	public void setWaitTimeout(long waitTimeout) {
		this.waitTimeout = waitTimeout;
	}
	
	public void start() {
		doSendPending();
	}
	
	/* (non-Javadoc)
	 * @see com.skymobi.transport.Endpoint#stop()
	 */
	public void stop() {
		super.stop();
		
        this.exec.shutdownNow();
	}
}
